-- created by kumiler
-- on 2022/11/22 16:46
-- DESC

insert overwrite table spmi_dwd.dwd_spmi_composite_bill_base_dt partition (dt)
select id,
       bill_no,
       send_financial_center_id,
       send_financial_center_name,
       send_financial_center_code,
       send_network_id,
       send_network_code,
       send_network_name,
       waybill_no,
       customer_code,
       customer_name,
       sign_financial_center_id,
       sign_financial_center_code,
       sign_financial_center_name,
       sign_network_id,
       sign_network_code,
       sign_network_name,
       receiver_province_id,
       receiver_province_name,
       franchiser_id,
       franchiser_name,
       sign_time,
       is_sign,
       piece_fee,
       transfer_fee,
       operation_fee,
       sales_price,
       base_dispatch_fee,
       receivable_fee,
       pass_price,
       avg_deduction_fee,
       diff_fee,
       sum_fee,
       customer_fee,
       piece_financial_center_id,
       piece_financial_center_code,
       piece_financial_center_name,
       piece_network_id,
       piece_network_code,
       piece_network_name,
       destination_id,
       destination_code,
       destination_name,
       cost_weight,
       avg_weight,
       settle_weight,
       product_type_id,
       product_type_name,
       product_type_code,
       transport_type_id,
       transport_type_name,
       transport_type_code,
       goods_type_id,
       goods_type_name,
       goods_type_code,
       is_bad_piece,
       is_refund,
       operation_network_id,
       operation_network_code,
       operation_network_name,
       start_financial_center_id,
       start_financial_center_code,
       start_financial_center_name,
       is_lock,
       lock_time,
       remarks,
       piece_version,
       version,
       is_delete,
       bill_generation_date,
       waybill_time,
       create_time,
       update_by,
       update_by_name,
       update_time,
       update_by_network_code,
       update_by_network_name,
       last_update_time_sync,
       last_update_time_piece,
       last_update_time_opt,
       last_update_time_tof,
       trace_id,
       received_total,
       is_bubble,
       cost_list_name,
       package_charge_weight_source,
       entry_mark,
       retrieve_status,
       entry_mark_time,
       center_package_fee,
       bill_generation_dt as dt
from (
         select *, row_number() over (partition by id,waybill_no order by update_time desc ) as bill_rank
         from (
                  select *, cast(to_date(waybill_time) as string) as bill_generation_dt
                  from spmi_ods.spmi_composite_bill
                  where dt > date_add('{{ execution_date | cst_ds }}', -90)
                    and dt <= '{{ execution_date | cst_ds }}'
                    and to_date(waybill_time) > date_add('{{ execution_date | cst_ds }}', -90)
                    and to_date(waybill_time) <= '{{ execution_date | cst_ds }}'
              ) t1
     ) t2
where bill_rank = 1
distribute by dt, abs(hash(waybill_no)) % 80;
